package com.crazymaker.springcloud.websocket.netty;

import com.crazymaker.springcloud.standard.context.SpringContextUtil;
import com.crazymaker.springcloud.websocket.netty.handler.ClientTokenCheckHandler;
import com.crazymaker.springcloud.websocket.netty.handler.PusMsgWebSocketFrameHandler;
import com.crazymaker.springcloud.websocket.netty.handler.ServerExceptionHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * Netty 服务
 * Created by 尼恩 @ 疯狂创客圈
 */
@Component
@Slf4j
public class WebSocketServer implements ApplicationContextAware {


    @Value("${tunnel.websocket.port}")
    private int websocketPort;


    @Value("${websocket.register.gateway}")
    private String websocketRegisterGateway;


    @Value("${websocket.thread.worker}")
    private String workerThreads;


    @Value("${websocket.thread.nio}")
    private String nioThreads;


    @Value("${websocket.thread.biz}")
    private String bizThreads;


    private Channel channel;
    private EventLoopGroup eventLoopGroupBoss;
    private EventLoopGroup eventLoopGroupNIO;
    private DefaultEventExecutorGroup workerExecuteGroup;
    private DefaultEventExecutorGroup bizExecutorGroup;


    private boolean useEpoll() {
        boolean epollReady = RemotingUtil.isLinuxPlatform()
                && Epoll.isAvailable();

        log.info("\n----------------------------------------------------------\n\t" +
                "epollReady is :{}\n\t", epollReady);
        return epollReady;
    }

    /**
     * 停止即时通讯服务器
     */
    public void stopServer() {
        if (channel != null) {
            channel.close();
        }
        eventLoopGroupBoss.shutdownGracefully();
        eventLoopGroupNIO.shutdownGracefully();
        workerExecuteGroup.shutdownGracefully();
        bizExecutorGroup.shutdownGracefully();
    }

    /**
     * 启动即时通讯服务器
     */
    public void startServer(int port) {

        createThreadPool();

        ChannelFuture channelFuture = null;
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupNIO)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .childHandler(new PushServerInitializer());
        InetSocketAddress address = new InetSocketAddress(port);
        channelFuture = bootstrap.bind(address);
//        channelFuture.syncUninterruptibly();

        channel = channelFuture.channel();
        // 返回与当前Java应用程序关联的运行时对象
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                stopServer();
            }
        });

        log.info("\n----------------------------------------------------------\n\t" +
                "Nett WebSocket 服务 is running! Access Port:{}\n\t", websocketPort);

        channelFuture.channel().closeFuture().syncUninterruptibly();
    }

    private void createThreadPool() {
        if (useEpoll()) {
            this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyNIOBoss_%d", this.threadIndex.incrementAndGet()));
                }
            });
            this.eventLoopGroupNIO = new EpollEventLoopGroup(Integer.parseInt(nioThreads), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettNIOWorker_%d", this.threadIndex.incrementAndGet()));
                }
            });
        } else {
            this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettyNIOBoss_%d", this.threadIndex.incrementAndGet()));
                }
            });
            this.eventLoopGroupNIO = new NioEventLoopGroup(Integer.parseInt(nioThreads), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("NettNIOWorker_%d", this.threadIndex.incrementAndGet()));
                }
            });
        }


        this.workerExecuteGroup = new DefaultEventExecutorGroup(Integer.parseInt(workerThreads), new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettNIOWorker_%d", this.threadIndex.incrementAndGet()));
            }
        });

        this.bizExecutorGroup = new DefaultEventExecutorGroup(
                Integer.parseInt(bizThreads),
                new ThreadFactory() {

                    private AtomicInteger threadIndex = new AtomicInteger(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "NettybizThreads_" + this.threadIndex.incrementAndGet());
                    }
                });
    }

    /**
     * 内部类
     */
    class PushServerInitializer extends ChannelInitializer<Channel> {
        private static final int READ_IDLE_TIME_OUT = 600; // 读超时  s
        private static final int WRITE_IDLE_TIME_OUT = 0;// 写超时
        private static final int ALL_IDLE_TIME_OUT = 0; // 所有超时


        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            // Netty自己的http解码器和编码器，报文级别 HTTP请求的解码和编码
            pipeline.addLast(workerExecuteGroup, new HttpServerCodec());
            // ChunkedWriteHandler 是用于大数据的分区传输
            // 主要用于处理大数据流，比如一个1G大小的文件如果你直接传输肯定会撑暴jvm内存的;
            // 增加之后就不用考虑这个问题了
            pipeline.addLast(workerExecuteGroup, new ChunkedWriteHandler());
            // HttpObjectAggregator 是完全的解析Http消息体请求用的
            // 把多个消息转换为一个单一的完全FullHttpRequest或是FullHttpResponse，
            // 原因是HTTP解码器会在每个HTTP消息中生成多个消息对象HttpRequest/HttpResponse,HttpContent,LastHttpContent
            pipeline.addLast(workerExecuteGroup, new HttpObjectAggregator(64 * 1024));
            pipeline.addLast(bizExecutorGroup, new ClientTokenCheckHandler());
            // WebSocket数据压缩
            pipeline.addLast(workerExecuteGroup, new WebSocketServerCompressionHandler());
            // WebSocketServerProtocolHandler是配置websocket的监听地址/协议包长度限制
            pipeline.addLast(workerExecuteGroup, new WebSocketServerProtocolHandler("/push", null, true, 10 * 1024));

            //当连接在60秒内没有接收到消息时，就会触发一个 IdleStateEvent 事件，
            // 此事件被 HeartbeatHandler 的 userEventTriggered 方法处理到
            pipeline.addLast(workerExecuteGroup, new IdleStateHandler(READ_IDLE_TIME_OUT, WRITE_IDLE_TIME_OUT, ALL_IDLE_TIME_OUT, TimeUnit.SECONDS));

            //WebSocketServerHandler、TextWebSocketFrameHandler 是自定义逻辑处理器，
            pipeline.addLast(bizExecutorGroup, new PusMsgWebSocketFrameHandler());
            pipeline.addLast(bizExecutorGroup, new ServerExceptionHandler());

        }
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContextUtil.setContext(applicationContext);
        new Thread(() -> {
            startServer(websocketPort);
        }).start();


    }
}
